package pulsar.subs;

import org.apache.pulsar.client.api.*;

import java.util.concurrent.TimeUnit;

/**
 * 
 * @version 1.0
 * @date 2021/3/24 17:26
 */
public class MyKeyProducer {

    static PulsarClient client;

    static Producer<byte[]> producer;

    static {
        try {
            client = PulsarClient.builder()
                        .serviceUrl(PulsarProperty.serviceUrl)
                        .build();
        } catch (PulsarClientException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws PulsarClientException, InterruptedException {
        sendMsg();

        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            try {
                client.close();
                client.close();
            } catch (PulsarClientException e) {
                e.printStackTrace();
            }
        }));
    }

    @SuppressWarnings("all")
    static void sendMsg() throws PulsarClientException, InterruptedException {
        producer = client.newProducer(Schema.BYTES)
                .topic(PulsarProperty.topic)
                //是否开启批量处理消息，默认true,需要注意的是enableBatching只在异步发送sendAsync生效，同步发送send失效。
                // 因此建议生产环境若想使用批处理，则需使用异步发送，或者多线程同步发送
                .enableBatching(false)
                //消息压缩（四种压缩方式：LZ4，ZLIB，ZSTD，SNAPPY），consumer端不用做改动就能消费，
//                // 开启后大约可以降低3/4带宽消耗和存储（官方测试）
//                .compressionType(CompressionType.LZ4)
//                //设置将对发送的消息进行批处理的时间段,10ms；可以理解为若该时间段内批处理成功，则一个batch中的消息数量不会被该参数所影响。
//                .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS)
                //设置发送超时0s；如果在sendTimeout过期之前服务器没有确认消息，则会发生错误。默认30s，设置为0代表无限制，建议配置为0
                .sendTimeout(0, TimeUnit.SECONDS)
//                //批处理中允许的最大消息数。默认1000
//                .batchingMaxMessages(1000)
//                //设置等待接受来自broker确认消息的队列的最大大小，默认1000
//                .maxPendingMessages(1000)
                //设置当消息队列中等待的消息已满时，Producer.send 和 Producer.sendAsync 是否应该block阻塞。
                // 默认为false，达到maxPendingMessages后send操作会报错，设置为true后，send操作阻塞但是不报错。建议设置为true
                .blockIfQueueFull(true)
//                //向不同partition分发消息的切换频率，默认10ms，可根据batch情况灵活调整
//                .roundRobinRouterBatchingPartitionSwitchFrequency(10)
//                //key_Shared模式要用KEY_BASED,才能保证同一个key的message在一个batch里
//                .batcherBuilder(BatcherBuilder.KEY_BASED)
                .create();
        producer.send("hello pulsar from bytes".getBytes());

        int seq = 1;
        while (true) {
            producer.flush();

            int key = (int) (Math.random() * 10 % 3) + 1;
            String keyMsg = "KEY-" + key;
            String msg = "got message! key:" + keyMsg + "; seq:" + seq;
            producer.newMessage()
                    .key(keyMsg)
                    .value(msg.getBytes())
                    .property("my-key", "my-value")
                    .property("my-other-key", "my-other-value")
                    .send();

            seq ++;
            Thread.sleep(1000);
        }
    }
}
